package com.heima.article.listener;



import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.checkerframework.checker.units.qual.K;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
/**
 * @BelongsProject: heima-leadnews
 * @BelongsPackage: com.heima.article.listener
 * @Author: wangjian
 * @CreateTime: 2022-07-08  17:44
 * @Description: TODO
 * @Version: 1.0
 */
@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {


    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {

        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Long articleId = updateArticleMessage.getArticleId();
                return new KeyValue<>(articleId.toString(), value);
            }
        });


        KGroupedStream<String, String> groupedStream = map.groupByKey();//根据可以分组



        TimeWindowedKStream<String, String> windowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(60))); //每60秒统计一次

//        聚合之前的初始化器 内容为空
        Initializer<String> initializer = new Initializer<String>() {
            @Override
            public String apply() {
                return null;
            }
        };

//        开始聚合
        Aggregator<String,String, String> aggregator = new Aggregator<String, String, String>() {
            @Override  //key:文章id
            public String apply(String key, String value, String aggregate) {
                ArticleStreamMessage articleStreamMessage  = null;
                if(aggregate==null){
                    articleStreamMessage = new ArticleStreamMessage();
                    articleStreamMessage.setArticleId(Long.parseLong(key));
                    articleStreamMessage.setView(0);
                    articleStreamMessage.setLike(0);
                    articleStreamMessage.setComment(0);
                    articleStreamMessage.setCollect(0);
                }else{
                    articleStreamMessage = JSON.parseObject(aggregate,ArticleStreamMessage.class);
                }

                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Integer type = updateArticleMessage.getType();
                switch (type){
                    case 0:  //阅读
                        articleStreamMessage.setView(articleStreamMessage.getView()+updateArticleMessage.getAdd());
                        break;
                    case 1: //点赞
                        articleStreamMessage.setLike(articleStreamMessage.getLike()+updateArticleMessage.getAdd());
                        break;
                    case 2: //评论
                        articleStreamMessage.setComment(articleStreamMessage.getComment()+updateArticleMessage.getAdd());
                        break;
                    case 3: //收藏
                        articleStreamMessage.setCollect(articleStreamMessage.getCollect()+updateArticleMessage.getAdd());
                        break;
                }
                return JSON.toJSONString(articleStreamMessage);
            }
        };
        KTable<Windowed<String>, String> kTable = windowedKStream.aggregate(initializer, aggregator);


        KStream<String, String> stream = kTable.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });

        return stream;


    }

}

